package com.example.datastreamdemo

import java.util

import akka.remote.WireFormats.TimeUnit
import org.apache.flink.streaming.api.datastream.DataStreamUtils
import org.apache.flink.streaming.api.environment.LocalStreamEnvironment
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

object iterateDemo {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val someIntegers: DataStream[Long] = env.generateSequence(0, 100000)


    val iteratedStream = someIntegers.iterate(
      iteration => {
        val minusOne = iteration.map(v => v - 1)
        val stillGreaterThanZero = minusOne.filter(_ > 0)
        val lessThanZero = minusOne.filter(_ <= 0)
        (stillGreaterThanZero, lessThanZero)
      }
    )

    iteratedStream.print()

    val value: util.Iterator[Long] = DataStreamUtils.collect(iteratedStream.javaStream)
    value.forEachRemaining(item => {
      println(item)
    })

    env.execute("afafa")

  }


//  private val env: StreamExecutionEnvironment = StreamExecutionEnvironment.createLocalEnvironment()
//
//  env.setBufferTimeout(Time.seconds(5).toMilliseconds)
//  env.generateSequence(1,10).setBufferTimeout(Time.seconds(5).toMilliseconds)
//  为了最大限度地提高吞吐量，请设置 setBufferTimeout(-1) 将删除超时，
//  缓冲区将仅在已满时刷新。若要最大程度地减少延迟，请将超时设置为接近 0 的值（例如 5 或 10 毫秒）。应避免缓冲区超时为 0，因为它可能会导致严重的性能下降。
}
